package com.lizhiyu.flink.demo8_cep;

import com.lizhiyu.flink.model.TimeUtils;
import com.mysql.cj.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * CEP 为复杂的事件处理  Complex event processing
 *
 * 在使用Flink时，对数据进行复杂的过滤，可以使用CEP来处理，CEP就相当于与把复杂的过滤条件简化成了个正则表达式
 * 通过数据根据CEP的匹配校验来判断数据是否符合要求
 *
 * CEP使用分3步
 * 1、定义Pattern
 * 2、数据源和 Pattern进行匹配
 * 3、获得匹配的数据
 *
 * 场景:例如:5秒内出现两次的数据
 *
 * 模式（Pattern）分为多种模式
 *      个体模式(Individual Patterns)：组成复杂规则的每一个单独的模式定义，就是个体模式
 *      组合模式(Combining Patterns)：很多个体模式组合起来，形成组合模式
 *      模式组(Groups of Patterns)：将一个组合模式作为条件嵌套在个体模式里，就是模式组
 */
public class ComplexEventProcess {

    //测试数据
//        小D,2022-11-11 12:01:01,-1
//        老王,2022-11-11 12:01:10,-1
//        老王,2022-11-11 12:01:11,-1
//        小D,2022-11-11 12:01:13,-1
//        老王,2022-11-11 12:01:14,-1
//        老王,2022-11-11 12:01:15,1
//        小D,2022-11-11 12:01:16,-1
//        老王,2022-11-11 12:01:17,-1
//        小D,2022-11-11 12:01:20,1
    /**
     * 实现每5秒钟连续登陆失败两次的用户
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {

        //构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> ds = env.socketTextStream("127.0.0.1", 8888);
        //用户名，时间，状态码 (状态码为负1则代表没有登陆成功)

        //string 转成tuple3
        DataStream<Tuple3<String, String, Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] arr = value.split(",");
                out.collect(Tuple3.of(arr[0], arr[1], Integer.parseInt(arr[2])));
            }
        });
        //设置窗口，指定 eventtime，使用顺序时间
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermakerDS = flatMapDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                //延迟策略去掉了延迟时间,时间是单调递增，event中的时间戳充当了水印
                .<Tuple3<String, String, Integer>>forMonotonousTimestamps()
                //生成一个延迟3s的固定水印
                //.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (event, timestamp) -> {
                            //指定POJO的事件时间列
                            return TimeUtils.strToDate(event.f1).getTime();
                        }
                ));
        //根据用户名分组
        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = watermakerDS.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        });
        //定义模式
        Pattern<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>> pattern = Pattern
                .<Tuple3<String, String, Integer>>begin("firstTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
            @Override
            public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                // -1 是登录失败错误码
                return value.f2 == -1;
            }
        })//.times(2).within(Time.seconds(10));//不是严格近邻
                .next("secondTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f2 == -1;
                    }
                }).within(Time.seconds(5));
        //匹配检查
        PatternStream<Tuple3<String, String, Integer>> patternStream = CEP.pattern(keyedStream, pattern);
        //获得匹配成功的结果
        SingleOutputStreamOperator<Tuple3<String, String,String>> select = patternStream.select(new PatternSelectFunction<Tuple3<String, String, Integer>, Tuple3<String, String,String>>() {
            @Override
            public Tuple3<String, String,String> select(Map<String, List<Tuple3<String, String, Integer>>> map) throws Exception {
                Tuple3<String, String, Integer> firstLoginFail =  map.get("firstTimeLogin").get(0);
                Tuple3<String, String, Integer> secondLoginFail =  map.get("secondTimeLogin").get(0);
                return Tuple3.of(firstLoginFail.f0,firstLoginFail.f1,secondLoginFail.f1);
            }
        });
        select.print("匹配结果");
        env.execute("CEP job");

    }
}
